package org.tity.commons.concurrent.workshop;

import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.tity.commons.concurrent.ResponsiveTask;
import org.tity.commons.concurrent.ThreadStopWay;
import org.tity.commons.concurrent.ThreadUtil;
import org.tity.commons.concurrent.pool.SimpleTaskManager;
import org.tity.commons.concurrent.pool.TaskManager;
import org.tity.commons.log.Logger;

/**
 * @author 李方捷
 * 生产者消费者模型，但队列
 * 
 * */
public class TaskWorkshop<K,V> {

	private int producerCount=1;
	private Class<? extends TaskManager> producerManagerType=SimpleTaskManager.class;
	private Producer<K,V> producer=null;
	private int producerTimeout=5000;
	private TaskManager producerManager=null;
	private int producerFrequency=1000;
	
	
	private int consumerCount=1;
	private Class<? extends TaskManager> consumerManagerType=SimpleTaskManager.class;
	private Consumer<K,V> consumer=null;
	private int consumerTimeout=5000;
	private int consumerFrequency=1000;
	private TaskManager consumerManager=null;
	private ThreadStopWay consumerStopWay=ThreadStopWay.INTERRUPT;
	
	public ThreadStopWay getConsumerStopWay() {
		return consumerStopWay;
	}



	public void setConsumerStopWay(ThreadStopWay consumerStopWay) {
		this.consumerStopWay = consumerStopWay;
	}


	private SimpleTaskManager workerMgr;
	
	private boolean started=false;
	
	private Queue<Element<K,V>> queue=null; 
	private ConcurrentHashMap<K,Byte> ids=new ConcurrentHashMap<>();
	
	 
	protected void offerElement(Element<K,V> e)
	{
		if(e==null) {
			return;
		}
		if(ids.contains(e.getId())) {
			return;
		}
		int i=0;
		while(true) {
			if(queue.offer(e) && i<128) {
				ids.put(e.getId(),Byte.MIN_VALUE);
				break;
			}
			ThreadUtil.sleep(0,64);
			i++;
		}
	}
	
	
	
	protected Element<K,V> pollElement()
	{
		Element<K,V> e=queue.poll();
		if(e!=null)
		{
			ids.remove(e.getId());
		}
		return e;
	}
	
	public int size()
	{
		if(queue==null) {
			return 0;
		}
		return queue.size();
	}
	
	
	public int getProducerCount() {
		return producerCount;
	}
	public Class<? extends TaskManager> getProducerManagerType() {
		return producerManagerType;
	}
	public Producer<K,V> getProducer() {
		return producer;
	}
	public int getConsumerCount() {
		return consumerCount;
	}
	public Class<? extends TaskManager> getConsumerManagerType() {
		return consumerManagerType;
	}
	public Consumer<K,V> getConsumer() {
		return consumer;
	}

	/**
	 * 设置生产者
	 * @param producer  生产者
	 * @param count    生产者数量
	 * */
	public void setProducer(Producer<K,V> producer,int count)
	{
		this.setProducer(null, producer, count,-1,-1);
	}

	/**
	 * 设置生产者
	 * @param producer  生产者
	 * @param count    生产者数量
	 * @param timeout  生产超时时间，毫秒
	 * @param frequency 生产周期，毫秒
	 * */
	public void setProducer(Producer<K,V> producer,int count,int timeout,int frequency)
	{
		this.setProducer(null, producer, count,timeout,frequency);
	}
	
	/**
	 * 设置生产者
	 * @param taskManagerType  生产者的任务管理器类型
	 * @param producer  生产者
	 * @param count    生产者数量
	 * @param timeout  生产超时时间，毫秒
	 * @param frequency 生产周期，毫秒
	 * */
	public void setProducer(Class<? extends TaskManager> taskManagerType,Producer<K,V> producer,int count,int timeout,int frequency)
	{
		if(taskManagerType!=null)
		{
			this.producerManagerType = taskManagerType;
		}
		if(count<=0) {
			count=1;
		}
		this.producerCount=count;
		if(timeout<=0) {
			timeout=5000;
		}
		this.producerTimeout=timeout;
		this.producer=producer;
		if(frequency<0) {
			frequency=1000;
		}
		this.producerFrequency=frequency;
	}
	
	/**
	 * 设置消费者
	 * @param consumer  消费者
	 * @param count 消费者数量
	 * */
	public void setConsumer(Consumer consumer,int count)
	{
		this.setConsumer(null, consumer, count,-1,-1);
	}
	
	/**
	 * 设置消费者
	 * @param consumer  消费者
	 * @param count 消费者数量
	 * @param timeout  消费超时时间，毫秒
	 * @param frequency 消费周期，毫秒
	 * */
	public void setConsumer(Consumer consumer,int count,int timeout,int frequency)
	{
		this.setConsumer(null, consumer, count,timeout,frequency);
	}
	
	/**
	 * 设置消费者
	 * @param taskManagerType  消费者的任务管理器类型
	 * @param consumer  消费者
	 * @param count 消费者数量
	 * @param timeout  消费超时时间，毫秒
	 * @param frequency 消费周期，毫秒
	 * */
	public void setConsumer(Class<? extends TaskManager> taskManagerType,Consumer<K,V> consumer,int count,int timeout,int frequency)
	{
		if(taskManagerType!=null)
		{
			this.consumerManagerType = taskManagerType;
		}
		if(count<=0) {
			count=1;
		}
		this.consumerCount=count;
		if(timeout<=0) {
			timeout=5000;
		}
		this.consumerTimeout=timeout;
		this.consumer=consumer;
		if(frequency<0) {
			frequency=1000;
		}
		this.consumerFrequency=frequency;
	}
	
	/**
	 * 开始工作
	 * */
	public synchronized void start()
	{
		if(started) {
			return;
		}
		
		if(producerCount==1) {
			if(consumerCount==1) {
				queue=new LinkedBlockingQueue<Element<K,V>>();
			} else {
				queue=new ConcurrentLinkedQueue<Element<K,V>>();
			}
		} else {
			if(consumerCount==1) {
				queue=new LinkedBlockingQueue<Element<K,V>>();
			} else {
				queue=new ConcurrentLinkedQueue<Element<K,V>>();
			}
		}
		
		if(queue==null) {
			queue=new ConcurrentLinkedQueue<Element<K,V>>();
		}
		
		workerMgr=new SimpleTaskManager(consumerCount*3);
		
		if(producerManager==null) {
			try {
				producerManager=producerManagerType.getConstructor(int.class).newInstance(producerCount*2);
			} catch (Exception e) {
				Logger.exception(e);
			}
		}
		
		if(consumerManager==null) {
			try {
				consumerManager=consumerManagerType.getConstructor(int.class).newInstance(consumerCount*2);
			} catch (Exception e) {
				Logger.exception(e);
			}
		}
		
		
		//启动生产者
		for (int i = 0; i < producerCount && producer!=null; i++) {
			producerManager.doIntervalTask(new Runnable() {
				@Override
				public void run() {
					List<Element<K,V>> list=producer.produce(size());
					if(list!=null) {
						for (Element<K, V> e : list) {
							offerElement(e);
						}
					}
				}
			}, producerFrequency);
		}
		
		//启动生产者
		for (int i = 0; i < consumerCount && consumer!=null; i++) {
			consumerManager.doIntervalTask(new Runnable() {
				@Override
				public void run() {
					Element<K, V> e=null;
					while(true) {
						e=pollElement();
						if(e!=null) {
							consume(e);
						}
					}
				}
			}, consumerFrequency);
		}
		
		started=true;
		
		Logger.info(this.getClass().getSimpleName()+" started");

	}


	private  class ConsumeTask implements Runnable
	{
		private Element<K, V> e;
		public ConsumeTask(Element<K, V> e)
		{
			this.e=e;
		}
		@Override
		public void run() {
			try {
				consumer.consume(this.e,size());
			} catch (Exception ex) {
				Logger.exception(ex);
			}
		}
	}
	

	protected void consume(Element<K, V> e) {
		
		ResponsiveTask task=new ResponsiveTask(workerMgr,new ConsumeTask(e), consumerTimeout,consumerStopWay);
		task.start();

	}
	
	
	
	
	
}
